package net.bw.realtime.game.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

/*
 * @ Author：liuyawei
 * @ Date：2025-06-13
 */
public class MongoDBJoinToKafkaJob {

    public static void main(String[] args) {

        // 1.创建表环境
        TableEnvironment tabEnv = getTableEnv();

        // 2. 输入表-input：映射到mongo消息队列
        createInputTable(tabEnv);

        // 3. 输出表-output：映射到kafka表
        createOutputTable(tabEnv) ;

        // 4. 保存数据-insert
        saveTokafka(tabEnv) ;

    }

    // todo 4.保存数据-insert
    private static void saveTokafka(TableEnvironment tabEnv) {

        //游戏维度表与游戏销售表关联，有数据更新则将所影响的销售表主键插入Kafka Topic
        tabEnv.executeSql(
                "INSERT INTO game_sales_fact (\n" +
                        "   sale_id\n" +
                        ")\n" +
                        "select\n" +
                        "  gs.sale_id\n" +
                        "from game_dimension as gd\n" +
                        "join game_sales FOR SYSTEM_TIME AS OF gd.process_time as gs\n" +
                        "on gd.game_id = gs.game_id"
        );

        //平台维度表与销售表关联，有数据更新则将所影响的销售主键插入Kafka事实表
        tabEnv.executeSql(
                "INSERT INTO game_sales_fact (\n" +
                        "   sale_id\n" +
                        ")\n" +
                        "select\n" +
                        "  gs.sale_id\n" +
                        "from platform_dimension as pd\n" +
                        "join game_sales FOR SYSTEM_TIME AS OF pd.process_time as gs\n" +
                        "on pd.platform_id = gs.platform_id"
        );

    }

    // todo 3.创建输出表
    private static void createOutputTable(TableEnvironment tabEnv) {

        // 创建Kafka Topic存储主键信息的事实表
        tabEnv.executeSql(
                "CREATE TABLE game_sales_fact (\n" +
                        "  sale_id      INT,\n" +
                        "  PRIMARY KEY (sale_id) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "  'connector' = 'upsert-kafka',\n" +
                        "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "  'topic' = 'game_sales_fact',\n" +
                        "  'key.format' = 'json',\n" +
                        "  'value.format' = 'json'\n" +
                        ");"
        );

    }

    // todo 2.创建输入表
    private static void createInputTable(TableEnvironment tabEnv) {

        //创建MongoDB游戏销售表
        tabEnv.executeSql("CREATE TABLE game_sales\n" +
                "(\n" +
                "  `_id`       STRING ,  \n" +
                "  sale_id     INT,      \n" +
                "  game_id     INT,      \n" +
                "  platform_id INT,      \n" +
                "  sale_date   STRING,   \n" +
                "  units_sold  INT,      \n" +
                "  sale_amt    Double,    \n" +
                "  process_time AS PROCTIME()  \n" +
                ")\n" +
                "WITH (\n" +
                " 'connector' = 'mongodb',\n" +
                " 'uri' = 'mongodb://localhost:27017',\n" +
                " 'database' = 'db',\n" +
                " 'collection' = 'game_sales'\n" +
                ")"
        );

        // 输出
        //tabEnv.executeSql("SELECT * FROM game_sales limit 20").print();

        // 创建游戏维度信息表
        tabEnv.executeSql("CREATE TABLE game_dimension\n" +
                "(\n" +
                "  `_id`          STRING,  \n" +
                "  game_name      STRING,  \n" +
                "  publisher      STRING,  \n" +
                "  developer      STRING,  \n" +
                "  release_date   STRING,  \n" +
                "  game_id        INT,     \n" +
                "  process_time AS PROCTIME()  \n" +
                ")\n" +
                "WITH (\n" +
                " 'connector' = 'mongodb',\n" +
                " 'uri' = 'mongodb://localhost:27017',\n" +
                " 'database' = 'db',\n" +
                " 'collection' = 'game_dimension'\n" +
                ")"
        );

        // 输出
        //tabEnv.executeSql("SELECT * FROM game_dimension limit 20").print();

        // 创建平台维度表
        tabEnv.executeSql(
              "CREATE TABLE platform_dimension(\n" +
                " `_id`          STRING, \n" +
                "  platform_id   INT,    \n" +
                "  type          STRING, \n" +
                "  platform_name STRING, \n" +
                "  process_time AS PROCTIME()  \n" +
                ")\n" +
                "WITH (\n" +
                " 'connector' = 'mongodb',\n" +
                " 'uri' = 'mongodb://localhost:27017',\n" +
                " 'database' = 'db',\n" +
                " 'collection' = 'platform_dimension'\n" +
                ")"
        );

        // 输出
        //tabEnv.executeSql("SELECT * FROM platform_dimension limit 20").print();

    }

    // todo 1.创建表环境
    private static TableEnvironment getTableEnv() {

        // 1环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();

        TableEnvironment tabEnv = TableEnvironment.create(settings) ;

        // 2配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");

        // 3返回对象
        return tabEnv;

    }

}
